package com.zx.learn.flink.source.custom;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * ：多并行度的自定义数据源
 * 自定义数据源, 每1秒钟随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
 */
@Slf4j
public class ParallelismDemo {
    public static void main(String[] args) throws Exception {
        // 1）获取flink流处理的运行环境
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8081);//设置webui的端口号
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        log.debug("执行环境默认的并行度：" + env.getParallelism());
        // 2）接入自定义数据源
        //The parallelism of non parallel operator must be 1.
        DataStreamSource<OrderBean> streamSource = env.addSource(new MySelfDefineSource()).setParallelism(2);
        //DataStreamSource<Order> streamSource = env.addSource(new MySource());
        log.debug("streamSource的并行度：" + streamSource.getParallelism());
        // 3）打印输出
        streamSource.print();

        // 4）启动运行作业
        env.execute();
    }
}
